/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ocr.mpp;

import cn.ac.iie.di.ocr.util.GlobalParameters;
import cn.ac.iie.di.ocr.util.MppJdbcPool_dbcp;
import java.io.File;
import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import javafx.scene.input.KeyCode;
import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.binary.Hex;
import org.codehaus.jackson.map.ObjectMapper;
import static cn.ac.iie.di.ocr.util.GlobalParameters.sendMsgQueue;

/**
 *
 * @author Austin
 */
public class SelectMppThread implements Runnable {

    static org.apache.log4j.Logger logger = null;

    static {

        logger = org.apache.log4j.Logger.getLogger(SelectMppThread.class.getName());

    }

    @Override
    public void run() {
        logger.info("UpdateMppThread is running");
        Boolean retry = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        while (true) {
            Connection mppcon = null;
            Statement sele = null;
            ArrayList<Map> jsonList = null;
            try {
                jsonList = new ArrayList<>(20);
                for (int i = 0; i < 20; ++i) {
                    Map str = GlobalParameters.mppQueue.poll(500, TimeUnit.MILLISECONDS);
                    if (null == str) {
                        logger.info("mppQueue is null ");
//                        Thread.sleep(1000);
                        break;
                    }

                    jsonList.add(str);
//                    logger.info("strstrstrstrstrstrstr" + str.toString());
                }
                logger.info("GlobalParameters.mppQueue.size()" + GlobalParameters.mppQueue.size());
                if (jsonList.size() == 0) {
                    logger.debug("updateQueue is null" + jsonList.toString() + " " + jsonList.size());
                    Thread.sleep(3000);
                    continue;
                }
                logger.info("the jsonList.size: " + jsonList.size());
                Long start = System.currentTimeMillis();
                for (int j = 0; j < 10; j++) {
                    try {
                        mppcon = MppJdbcPool_dbcp.getConnection();
                        if (mppcon != null) {
                            break;
                        }
                        Thread.sleep(3000);
                    } catch (Exception e) {
                        logger.error("mpp num " + j + " error " + e, e);
                        Thread.sleep(3000);
                    }
                }
                logger.info("get mppcon ok ");
                if (null == mppcon) {
                    logger.info("mppcon is null ");
                    for (Map map : jsonList) {
                        GlobalParameters.sendRetryQueue.put(map);
                    }
                    continue;
                }
                int count = 0;
                logger.debug("select startup! jsonList: " + jsonList.toString());
                for (Map map : jsonList) {
                    String sql = null;
                    Long stime = null;
                    try {
                        logger.debug("select map is : " + map.toString());
//                        Long time = Long.valueOf(map.get("starttime").toString());
//                        if ((System.currentTimeMillis() / 1000 - time) > 172800) {
//                            logger.info("the opid select all mpp is null userid: " + map.get("userid") + " ct: " + map.get("ct") + " groupid: " + map.get("groupid"));
//                            continue;
//                        }
//                        if (map.containsKey("lasttime")) {
//                            if (System.currentTimeMillis() - Long.valueOf(map.get("lasttime").toString()) > 900000) {
//                                logger.debug("lasttime is : " + Long.valueOf(map.get("lasttime").toString()) + "_" + (System.currentTimeMillis() - Long.valueOf(map.get("lasttime").toString())));
//                                GlobalParameters.sendRetryQueue.put(map);
//                                continue;
//                            }
//                        } else {
//                            map.put("lasttime", System.currentTimeMillis());
//                        }

                        stime = System.currentTimeMillis();
                        sql = "select m_type,m_content,u_name,u_send_ip,u_loc_county,u_loc_province,m_mm_id,m_mm_feature,m_dom_for,m_country_code,m_mm_url from " + GlobalParameters.SELECT_TABLE_NAME + " where u_ch_id=" + map.get("userid") + " and m_chat_room=" + map.get("groupid") + " and m_ch_id='" + map.get("ct") + "';";
//                    String sql = "select m_type,m_content,u_name,u_send_ip,u_loc_county,u_loc_province,m_mm_id,m_mm_feature,m_dom_for,m_country_code,m_mm_url from tp_wxq_entire where u_ch_id=" + map.get("userid") + " and m_chat_room=" + map.get("groupid") + " and m_ch_id='" + 000 + "';";
                        logger.info("the mpp select Sql is : " + sql);
                    } catch (Exception e) {
                        logger.error("splitor error: " + e, e);
                    }

                    while (true) {
                        sele = mppcon.createStatement();
                        if (sele != null) {
                            break;
                        }
                        logger.info("createStatement is null");
                    }
                    logger.info("createStatement success! ");
                    for (int i = 0; i < 30; i++) {
                        try {
                            rs = sele.executeQuery(sql);
                            if (rs != null) {
                                break;
                            }
                            logger.info("executeQuery ResultSet is null");
                            Thread.sleep(3000);
                        } catch (Exception e) {
                            logger.error("get executeQuery num: " + i + "; " + e, e);
                        }
                    }
                    logger.info("executeQuery the sql success! ");
                    if (null == rs) {
                        try {
                            rs = sele.executeQuery(sql);
                        } catch (Exception e) {
                            logger.error("implement sql error: " + e, e);
                            GlobalParameters.sendRetryQueue.put(map);
                        }
                    }
                    try {
                        logger.info("ResultSet " + rs.getMetaData().getColumnCount());
                    } catch (Exception e) {
                        logger.error(e, e);
                        GlobalParameters.sendRetryQueue.put(map);
                        continue;
                    }

                    List<Map> update = new ArrayList();

                    try {
                        while (rs.next()) {
                            Map result = new HashMap<>();
                            result.put("type", rs.getInt("m_type"));
                            if (null != rs.getString("m_content")) {
                                result.put("cont", rs.getString("m_content"));
                            } else {
                                result.put("cont", "");
                            }
                            result.put("uwxid", rs.getString("u_name"));
                            result.put("usip", rs.getString("u_send_ip"));
                            result.put("ulc", rs.getInt("u_loc_county"));
                            result.put("ulp", rs.getInt("u_loc_province"));
                            result.put("mmid", rs.getLong("m_mm_id"));
                            if (null != rs.getString("m_mm_feature")) {
                                String mm = rs.getString("m_mm_feature");
                                try {
                                    result.put("mmf", mm.substring(mm.indexOf("{"), mm.indexOf("}") + 1).replaceAll("\\\\", ""));
                                } catch (Exception e) {
                                    logger.error("error " + mm, e);
                                    result.put("mmf", "");
                                }

                            } else {
                                result.put("mmf", "");
                            }
                            result.put("mdf", rs.getInt("m_dom_for"));
                            result.put("mccode", rs.getInt("m_country_code"));
                            if (null != rs.getString("m_mm_url")) {
                                result.put("murl", rs.getString("m_mm_url"));
                            } else {
                                result.put("murl", "");
                            }
                            result.put("uchid", map.get("userid").toString());
                            result.put("mcr", map.get("groupid").toString());
                            result.put("mid", map.get("ct").toString());
                            logger.info("result " + result.toString());
                            update.add(result);
                        }
                    } catch (Exception e) {
                        logger.error(e, e);
                        GlobalParameters.sendRetryQueue.put(map);
                    }

                    logger.info("the result put success! ");
                    if (rs != null) {
                        rs.close();
                    }
                    if (sele != null) {
                        sele.close();
                    }
                    logger.info(" select all mpp time " + (System.currentTimeMillis() - stime));
                    if (update.size() == 0) {
                        try {
                            map.put("lasttime", System.currentTimeMillis());
                            logger.info("no select content");
                            String updatesql = "update " + GlobalParameters.UPDATE_TABLE_NAME + " set m_recall_count=cast(m_recall_count+1 as int) where u_ch_id=" + map.get("userid") + " and m_chat_room=" + map.get("groupid") + " and m_ch_id='" + map.get("ct") + "';";
                            logger.info("update sql: " + updatesql);
                            ps = mppcon.prepareStatement(updatesql);
                            ps.executeUpdate();
                            logger.info("update suess");

                        } catch (Exception e) {
                            logger.error("update error: " + e, e);
                        } finally {
                            if (ps != null) {
                                ps.close();
                            }
                            GlobalParameters.sendRetryQueue.put(map);
                            logger.info("update give retry");
                            continue;
                        }
                    }
                    count++;
                    sendMsgQueue.put(update);
                    logger.info("sendMsgQueue " + update.toString());

                }

                logger.info("select mpp sucess time: " + (System.currentTimeMillis() - start) + " ms count num: " + count);
                if (ps != null) {
                    ps.close();
                }
                if (rs != null) {
                    rs.close();
                }
                if (sele != null) {
                    sele.close();
                }
                MppJdbcPool_dbcp.removeConn(mppcon);

            } catch (Exception e) {
                logger.error("mpp update error " + e, e);
            } finally {
                if (mppcon != null) {
                    MppJdbcPool_dbcp.removeConn(mppcon);
                }
                if (ps != null) {
                    try {
                        ps.close();
                    } catch (SQLException ex) {
                        logger.error(ex, ex);
                    }
                }
                if (rs != null) {
                    try {
                        rs.close();
                    } catch (SQLException ex) {
                        logger.error(ex, ex);
                    }
                }
            }
        }

    }

    private static Map<String, String> getResultMap(ResultSet rs) throws SQLException {
        Map<String, String> hm = new HashMap<String, String>();
        ResultSetMetaData rsmd = rs.getMetaData();
        int count = rsmd.getColumnCount();
        logger.info("count " + count);
        for (int i = 1; i <= count; i++) {
            String key = rsmd.getColumnLabel(i);
            String value = rs.getString(i);
            hm.put(key, value);
        }
        return hm;
    }

}
